package com.xzx.flume;

import lombok.extern.slf4j.Slf4j;
import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;

/**
 * @author xinzhixuan
 * @version 1.0
 * @date 2025/5/11 22:54
 */
@Slf4j
public class MySink extends AbstractSink implements Configurable {
    private String name;

    @Override
    public void configure(Context context) {
        name = context.getString("name", "mySink");
    }

    @Override
    public Status process() throws EventDeliveryException {
        Status status;

        // Start transaction
        Channel ch = getChannel();
        Transaction txn = ch.getTransaction();
        txn.begin();
        try {
            // This try clause includes whatever Channel operations you want to do
            Event event = ch.take();
            while (event == null) {
                event = ch.take();
            }

            // Send the Event to the external repository.
            storeSomeData(event);

            txn.commit();
            status = Status.READY;
        } catch (Throwable t) {
            log.error("mySink error", t);
            txn.rollback();

            // Log exception, handle individual exceptions as needed

            status = Status.BACKOFF;

            // re-throw all Errors
            if (t instanceof Error) {
                throw (Error) t;
            }
        } finally {
            txn.close();
        }
        return status;
    }

    private void storeSomeData(Event event) {
        log.info("我的sink name:{} 获取到的数据 key：{}, value:{}", name, event.getHeaders(), new String(event.getBody()));
    }


}
